
package com.kweative.cluster;

import com.kweative.util.ByteUtil;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.net.DatagramPacket;
import java.net.MulticastSocket;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author henrik
 */
public class Receiver extends Transceiver {

    @Override
    public void run() {
        MulticastSocket s = Cluster.instance().getSocket();
        ObjectInputStream in = null;
        while(s.isBound()) {
            try {
                byte[] bytes = new byte[4028];
                s.receive(new DatagramPacket(bytes,bytes.length));
                in = new ClusterInputStream(new ByteArrayInputStream(bytes));
                
                ActionPackage ap = (ActionPackage) in.readObject();
                boolean closeLoop = false;
                boolean isSender = ap.getSender().equals(Cluster.node());
                for(Action a:ap) {
                    if (a.getReceiver() != null && !a.getReceiver().equals(Cluster.node())) {
                        //This is not the receiver..
                        continue;
                    }

                    switch(a.getType()) {
                        case DELETE:
                            //System.out.println("CLUSTER: DEL " + a.getKey());
                            Cluster.instance().getDatastore().remove(a.getKey());
                            break;
                        case UPDATE:
                            //System.out.println("CLUSTER: UPDATE " + a.getKey());
                        case INSERT:
                            //System.out.println("CLUSTER: INSERT " + a.getKey());
                            Cluster.instance().getDatastore().put(a.getKey(),a.getObject());
                            break;
                        case INIT:
                            //Called from master - to bring new node up to speed
                            Serializable[] data = (Serializable[]) a.getObject();;
                            HashMap<String, Serializable> datastore = (HashMap<String, Serializable>) data[0];
                            LinkedList<ClusterNode> nodes = (LinkedList<ClusterNode>) data[1];
                            
                            Cluster.instance().setDatastore(datastore);
                            Cluster.instance().setMaster(ap.getSender());
                            Cluster.instance().setNodes(nodes);
                            Cluster.node().setId(Cluster.nodes().size());
                            System.out.println(Cluster.node() + ": " + nodes.size() + " received");
                            Cluster.instance().setReady(true);
                            break;
                        case GREET:
                            if (isSender) break;
                            //New node sends its own ClusterNode object for remote registries.
                            ClusterNode newNode = (ClusterNode) a.getObject();
                            Cluster.nodes().push(newNode);
                            //Old nodes send their own to the new node
                            
                            if (Cluster.master() == null) {
                                boolean foundOlder = false;
                                for(ClusterNode n:Cluster.nodes()) {
                                    if (n.getCreated().before(Cluster.node().getCreated()))
                                        foundOlder = true;
                                }
                                if (!foundOlder) {
                                    //Found the master...
                                    Cluster.transmit(new Action(Action.ActionType.SETMASTER, Cluster.node(), null));
                                    Cluster.transmit(new Action(Action.ActionType.INIT,Cluster.datastore(), null, newNode));
                                }
                            } else if (Cluster.isMaster()) {
                                //If we hit the master - send all data to the new node
                                Cluster.transmit(new Action(Action.ActionType.INIT,new Serializable[]{Cluster.datastore(),Cluster.nodes()}, null, newNode));
                            }
                            break;
                        case SETMASTER:
                            ClusterNode master = (ClusterNode) a.getObject();
                            Cluster.instance().setMaster(master);
                            break;
                        case LEAVE:
                            Cluster.nodes().remove(ap.getSender());
                            if (Cluster.isMaster() && isSender) {
                                ClusterNode oldest = null;
                                for(ClusterNode n:Cluster.nodes()) {
                                    if (oldest == null)
                                        oldest = n;
                                    else if (n.getCreated().before(oldest.getCreated()))
                                        oldest = n;
                                }
                                if (oldest != null) {
                                    //Found the new master...
                                    Cluster.transmit(new Action(Action.ActionType.SETMASTER, oldest, null));
                                } else {
                                    System.out.println("CLUSTER: NO NODES LEFT??");
                                }
                            }
                            System.out.println("CLUSTER: NODE REMOVED");
                            //Cluster.store("nextId",Cluster.instance().getNodes().size());
                            break;
                    }
                }
                
            } catch (ClassNotFoundException ex) {
                Logger.getLogger(Receiver.class.getName()).log(Level.SEVERE, null, ex);
            } catch (IOException ex) {
                Logger.getLogger(Receiver.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}
